package com.stone.customer.console.job;

import com.stone.customer.console.infra.constant.CustomerConsoleConstant;
import com.stone.message.adapter.infra.publisher.MessagePublisher;
import com.stone.message.adapter.trigger.AbstractReceiverTrigger;
import com.stone.starter.redis.client.RedisCacheClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Map;

/**
 * <p>
 * 抽象job接收器，统一处理从Redis拉取数据的一类操作。子类需要从Redis不同的队列里拉取数据的话
 * 1、只需要继承AbstractReceiverFromRedisTriggerJob
 * 2、然后实现redisQueueName()方法：指定从Redis哪个队列获取数据
 * 3、实现messageQueueName()方法：指定将消息发送到哪个queue
 * 4、实现jobName()方法：指定job名称，方便在出错的时候做错误补偿
 * </p>
 * 注意：该job触发器，默认是同步执行，若需要多线程开启异步，则覆写isAsyncExecute()方法即可
 *
 * @author Mr_wenpan@163.com 2021/07/17 18:22
 */
@Slf4j
public abstract class AbstractReceiverFromRedisTriggerJob extends AbstractReceiverTrigger<String> {

    @Autowired
    private RedisCacheClient redisCacheClient;

    @Autowired
    @Qualifier("receiverFromRedisTriggerPool")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private static final String DEFAULT_THREAD_COUNT = "1";

    /**
     * job执行自动调用方法
     *
     * @author Mr_wenpan@163.com 2021/7/17 6:25 下午
     */
    public void autoExecuteJob(Map<String, String> params) {
        log.info("job名称：{}，job执行参数：{}", jobName(), params);
        String threadNum = params.get(CustomerConsoleConstant.JobExecuteReference.THREAD_NUM);
        threadNum = StringUtils.isBlank(threadNum) ? DEFAULT_THREAD_COUNT : threadNum;

        trigger(messageQueueName(), Integer.parseInt(threadNum));

        log.info("job执行完毕！job名称：{},job参数：{}", jobName(), params);
    }

    public AbstractReceiverFromRedisTriggerJob(MessagePublisher messagePublisher) {
        super(messagePublisher);
    }

    @Override
    protected ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        return threadPoolTaskExecutor;
    }

    @Override
    protected String getData() {
        // job的数据从Redis获取，也可以留给具体的子类实现
        return redisCacheClient.opsForList().rightPop(redisQueueName());
    }

    @Override
    protected String errorCompensate(String data) {
        // 对于从Redis拉数据的job统一做错误补偿
        log.info("[{}] job 执行出现异常，执行错误补偿，错误数据为：{}", jobName(), data);
        // 写入数据库job失败记录表，或重新放入Redis队列，或放入Redis中另一个指定的队列（如果需要放入另一个错误队列，这里需要指定错误队列名称）
        // todo

        log.info("[{}] job 执行出现异常，错误补偿完毕，错误数据为：{}", jobName(), data);
        return data;
    }

    /**
     * job名称
     *
     * @return java.lang.String
     * @author Mr_wenpan@163.com 2021/7/17 6:42 下午
     */
    protected abstract String jobName();

    /**
     * 消息队列名称
     *
     * @return java.lang.String
     * @author Mr_wenpan@163.com 2021/7/17 6:42 下午
     */
    protected abstract String messageQueueName();

    /**
     * 从Redis拉取数据的队列名
     *
     * @return java.lang.String
     * @author Mr_wenpan@163.com 2021/7/17 6:53 下午
     */
    protected abstract String redisQueueName();
}
